/*
 *  Licensed to the Apache Software Foundation (ASF) under one or more
 *  contributor license agreements.  See the NOTICE file distributed with
 *  this work for additional information regarding copyright ownership.
 *  The ASF licenses this file to You under the Apache License, Version 2.0
 *  (the "License"); you may not use this file except in compliance with
 *  the License.  You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 *  Unless required by applicable law or agreed to in writing, software
 *  distributed under the License is distributed on an "AS IS" BASIS,
 *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *  See the License for the specific language governing permissions and
 *  limitations under the License.
 */

// export CATALINA_BASE=$HOME/Documents/workspace
package websocket.proxy;

import java.io.IOException;
import java.nio.ByteBuffer;

import jakarta.websocket.OnMessage;
import jakarta.websocket.PongMessage;
import jakarta.websocket.Session;
import jakarta.websocket.server.ServerEndpoint;

// peter: use a hash map to store <Session, Socket> pair
import java.net.Socket;
import java.util.concurrent.ConcurrentHashMap;
import java.io.OutputStream;
import java.io.InputStream;


/**
 * The three annotated echo endpoints can be used to test with Autobahn and
 * the following command "wstest -m fuzzingclient -s servers.json". See the
 * Autobahn documentation for setup and general information.
 */
@ServerEndpoint("/websocket/proxy")
public class Proxy {

    static {
        System.out.println("Proxy static init");
        Class cls = null;
        try {
            cls = Class.forName("wapeter.server.Server");
        } catch (Exception e) {
            System.err.println("Proxy.static cannot find wapeter.server.Server");
        }
        if (cls == null) {
            System.err.println("Proxy.static cannot find Server cls=null");
        }
    }

    final static ConcurrentHashMap<Session, Socket> map = new ConcurrentHashMap<Session, Socket>();

    // ReadTCP thread wait for message from tcp socket 
    // then send back the message to websocket (via Session)
    class ReadTCP extends Thread
    {
        Session session;
        Socket sock;
        boolean running = true;

        ReadTCP(Session _session, Socket _sock) {
            session = _session;
            sock = _sock;
        }

        @Override
        public void run() {
            InputStream in;
            int n;
            byte [] byteList = new byte[1024]; // 100 enough ? 1024 safe

            try {
                in = sock.getInputStream();
            } catch (Exception e) {
                System.err.println("Proxy.ReadTCP.run getInputStream : ex" + e);
                return;  // no hope to revive
            }
            try {
                while ((n = in.read(byteList)) > 0) {
                    // TODO session. xxx. send
                    // what is last ?  true means last message? 
                    // false means more message pending?
                    String msg = new String(byteList, 0, n);
                    session.getBasicRemote().sendText(msg, true);
                }
            } catch (Exception e) {
                System.err.println("Proxy.ReadTCP.run read ex: " + e);
                try { sock.close(); } catch (Exception e2) {} // ignore
                map.remove(session);
            }

        }

        public void interrupt() {
            running = false;
            try {
                sock.close();
            } catch (Exception e) {
                System.err.println("Proxy.ReadTCP.interrupt close ex: " + e);
            }
        }

    }


    int sendTCP(Session session, String msg) {
        Socket sock;
        int ret = 0;

        sock = map.get(session);
        if (sock == null) {
            System.out.println("Proxy.sendTCP:  create new sock");
            try {
                sock = new Socket("127.0.0.1", 7749);
            } catch (Exception e) {
                System.err.println("Proxy.sendTCP ex: " + e);
                return -9; // access denied
            }
            // assume success here
            // create a new thread and put the map (session, socket)
            Socket oldSock;

            oldSock = map.put(session, sock);
            if (null != oldSock) {
                try {
                    oldSock.close();
                } catch (Exception e) {
                    // this error can be ignored
                    System.err.println("Proxy.sendTCP oldSock.close ex: " + e);
                }
            }
            // TODO create read thread
            new ReadTCP(session, sock).start();
        }


        // here: sock is not null
        ret = 0; // assume ok, out.write() does NOT return value
        try {
            OutputStream out = sock.getOutputStream();
            out.write(msg.getBytes()); // no linefeed, ok ?
            out.flush();
        } catch (Exception e) {
            System.err.println("Proxy.sendTCP write ex: " + e);
            ret = -5;
            try {
                sock.close();
            } catch (Exception e2) {
                System.err.println("Proxy.sendTCP write fail sock.close ex: " + e2);
            }
            map.remove(session); // usually disconnect, remove
        }

        return ret;
    }



    @OnMessage
    public void onMessage(Session session, String msg, boolean last) {
        int ret;
        System.out.println("app.Proxy.onMessage: " + msg);
        try {
            if (session.isOpen()) {
                ret = sendTCP(session, msg);
                if (ret < 0) {
                    session.getBasicRemote().sendText("ERROR app.Proxy.sendTCP " + ret + "\n");
                }
                // peter : do NOT send:
                // Proxy.ReadTCP.run read : exjava.lang.IllegalStateException: The remote endpoint was in state [TEXT_PARTIAL_WRITING] which is an invalid state for called method
                // session.getBasicRemote().sendText("Proxy.java sid=: " + session.getId() + "  msg=" + msg, last);
            }
        } catch (Exception e) {
            try {
                session.close();
            } catch (IOException e1) {
                // Ignore
            }
        }
    }

    @OnMessage
    public void onBinaryMessage(Session session, ByteBuffer bb,
            boolean last) {
        System.out.println("Proxy.onBinaryMessage: ");
        try {
            if (session.isOpen()) {
                session.getBasicRemote().sendBinary(bb, last);
            }
        } catch (IOException e) {
            try {
                session.close();
            } catch (IOException e1) {
                // Ignore
            }
        }
    }

    /**
     * Process a received pong. This is a NO-OP.
     *
     * @param pm    Ignored.
     */
    @OnMessage
    public void echoPongMessage(PongMessage pm) {
        // NO-OP
    }

    // XXX not support @OnError, why?
    // @OnError
    public void onError(Session session, Throwable t) {
        System.err.println("Proxy.onError session=" + session);
        if (null != t) {
            System.err.println("Proxy.onError throwable=" + t.getMessage());
            t.printStackTrace();
        }
    }
}
